package com.atguigu.edu.realtime.app.func;


import com.alibaba.fastjson.JSONObject;
import com.atguigu.edu.realtime.common.EduConfig;
import com.atguigu.edu.realtime.util.PhoenixUtil;
import org.apache.commons.lang3.StringUtils;
import org.apache.flink.streaming.api.functions.sink.SinkFunction;

/**
 * @Author zhangsan
 * @Date 2022/10/18 16:49
 * @Description //TODO  将流中的数据写到phoenix表中
 */
public class DimSinkFunction implements SinkFunction<JSONObject> {
    @Override
    public void invoke(JSONObject jsonObj, Context context) throws Exception {
        //获取向phoenix中写入数据的表名
        String sinkTable = jsonObj.getString("sink_table");
        //从jsonObj中过滤掉sink_table
        jsonObj.remove("sink_table");

        String type = jsonObj.getString("type");
        jsonObj.remove("type");

        //拼接upsert语句
        String upsertSql = "upsert into " + EduConfig.PHOENIX_SCHEMA + "." + sinkTable
                + "(" + StringUtils.join(jsonObj.keySet(), ",") + ") " +
                " values('" + StringUtils.join(jsonObj.values(), "','") + "')";
        //System.out.println("向phoenix表中插入数据的SQL:" + upsertSql);

        PhoenixUtil.executeSql(upsertSql);

        /*//如果对维度表数据进行了更新，需要将redis中缓存的数据删除掉
        if ("update".equals(type)){
            DimUtil.deleteCached(sinkTable,jsonObj.getString("id"));
        }*/

    }
}
